// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

/// There are 2 styles of position.
/// 1. LongShort (China)
///   long = 10  = net + buy  working order
///   short = 3  = net - sell worknig order
///    net =  7  = net = buy working order + sell working order

/// China-style
///                net = long_qty  -   short_qty
/// BUY_OPEN        +        +                   --v   <- max_long_qty chk
/// BUY_CLOSE       +                      -     --^   <- max_long_qty chk + how much can we close
///
/// SELL_OPEN       -                      +     --v   <- max_short_qty chk
/// SELL_CLOSE      -        -                   --^   <- max_short_qty chk + how much can we close
///
/// non-China-style
///
/// BUY             +                                  <- max_long_qty chk
/// SELL            -                                  <- max_short_qty chk
///
/// Hence,
/// OPEN_AND_CLOSE <- net chk (on total qty) + max_qty chk (on first qty) + net chk (on second qty)
/// + max_qty chk (on second qty)

#include <algorithm>
#include <functional>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <vector>
#include <deque>
#include <utility>
#include <unordered_map>
#include <spdlog/spdlog.h>

#include "coin2/base/config/app_config.h"
#include "coin2/base/log.h"
#include "coin2/exchange/base/executor/archive_executor.h"
#include "coin2/exchange/base/order/order.h"
#include "coin2/exchange/base/symbology/product_info.h"
#include "coin2/exchange/factory/symbology/product.h"

#include "coin/proto/coin_order_enums.pb.h"
#include "coin2/exchange/order/system.pb.h"
#include "coin2/strategy/order_executor.pb.h"
#include "coin2/strategy/order_executor/util.h"
#include "coin2/strategy/util/moving_window.h"
#include "coin2/strategy/order_executor/layering_executor.h"


namespace coin2::strategy::order_executor {

namespace impl {

using coin2::exchange::base::executor::ArchiveSyncExecutor;
using coin2::exchange::base::executor::LiveAsyncExecutor;
using coin2::exchange::base::order::OrderSpec;
using coin2::exchange::base::order::IOrderContext;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::base::symbology::IProductHolder;
using coin2::exchange::base::symbology::TickQty;
using coin::proto::OrderDirection;
using coin::proto::OrderDuration;
using coin::proto::OrderSide;
using coin::proto::OrderState;
using coin::proto::OrderType;
using coin2::strategy::ChinaStyleOrderSidePolicy;
using coin2::strategy::OrderExecutorConfig;
using coin2::strategy::OrderExecutorSystemConfig;
using exchange::base::order::IOrderSubsystem;
using exchange::base::order::OrderQuerier;

// struct AdjustedPrice {
//   double price() { return price_; }
//   AdjustedPrice NewCeil() {
//     THROW() << "not impl";
//     return AdjustedPrice(0);
//   }

//   AdjustedPrice NewFloor() {
//     THROW() << "not impl";
//     return AdjustedPrice(0);
//   }

//   AdjustedPrice NewExact(double price) {
//     THROW() << "not impl";
//     return AdjustedPrice(0);
//   }

//   AdjustedPrice NewUnsafe(double price) { return AdjustedPrice{price}; }

//  protected:
//   AdjustedPrice(double price) : price_(price) {}
//   double price_;
// };

struct OrderExecutionResult {
  // A strategy can check the full submission by
  // submitted_qty == qty

  bool ok() const { return qty > 0; }
  bool fail() const { return qty == 0; }
  int64_t first_oid = 0;
  int64_t second_oid = 0;
  double qty = 0;
};

struct SideQty {
  OrderSide side;
  double qty;
};

class OrderExecutor {
 public:
  using OrderUpdate = coin2::exchange::base::order::OrderUpdate;
  using ActionType = coin::proto::RequestType;

  int64_t last_timestamp_ = 0L;
  mutable int64_t lastlogt_fillpos_ = 0L;
  mutable int64_t lastlogt_pos_ = 0L;

  OrderExecutor(
      ExchangeType exchange,
      const std::string& name,
      const std::string& order_symbol,
      const OrderExecutorConfig& config,
      IOrderSubsystem* os);

  // not compatible with UpdateScale(). will be very confusing.
  void AdjustSize(
      double min_position,
      double max_position,
      double max_lot_size,
      std::optional<double> min_order_qty,
      std::optional<double> maintain_qty) {
    size_shape_configured_ = true;
    AdjustSizeImpl(
        min_position,
        max_position,
        max_lot_size,
        min_order_qty,
        maintain_qty);
  }

  void UpdateScale(double scale_multiplier) {
    CHECK(!size_shape_configured_);
    AdjustSizeImpl(
        scale_multiplier * config_orig_.qty().min_position(),
        scale_multiplier * config_orig_.qty().max_position(),
        scale_multiplier * config_orig_.qty().max_lot_size(),
        scale_multiplier * config_orig_.qty().min_order_qty(),
        scale_multiplier * config_orig_.side().maintain_qty());
    AdjustQuoteSizeImpl(
        scale_multiplier * config_orig_.qty().min_position_quote(),
        scale_multiplier * config_orig_.qty().max_position_quote());
  }

  void AdjustSizeImpl(
      double min_position,
      double max_position,
      double max_lot_size,
      std::optional<double> min_order_qty,
      std::optional<double> maintain_qty) {
    config_.mutable_qty()->set_min_position(min_position);
    config_.mutable_qty()->set_max_position(max_position);
    config_.mutable_qty()->set_max_lot_size(max_lot_size);
    if (min_order_qty) {
      config_.mutable_qty()->set_min_order_qty(*min_order_qty);
    }
    if (maintain_qty) {
      config_.mutable_side()->set_maintain_qty(*maintain_qty);
    }
    if (!config_.disable_verbose_logging())
      SPDLOG_INFO("Size Updated: {}={}", order_symbol_, config_.DebugString());
  }

  void AdjustQuoteSizeImpl(
      double min_position_quote,
      double max_position_quote) {
    config_.mutable_qty()->set_min_position_quote(min_position_quote);
    config_.mutable_qty()->set_max_position_quote(max_position_quote);
  }

  // og pos consolidation begin
  void onAccountInfo(
      const IProduct& product,
      const OrderUpdate& upd) {
    last_timestamp_ = upd.timestamp();
    if (os_->is_ready() && oq_->has_position_info(product)) {
      double og_pos = oq_->position(product);
      if (use_fill_pos_ && fill_pos_ == std::nullopt) {
        fill_pos_ = og_pos;
        SPDLOG_INFO(
            "{} Init {} fill_pos {}", upd.timestamp(), order_symbol_, *fill_pos_);
      }
      if (std::abs(og_pos - og_pos_) > 1.e-6) {
        og_pos_ = og_pos;
        og_pos_ts_ = upd.timestamp();
        SPDLOG_INFO(
            "{} Update {} og_pos {}", upd.timestamp(), order_symbol_, og_pos_);
      }
      if (upd.timestamp() - std::max(og_pos_ts_, last_fill_ts_) > update_pos_with_og_pos_period_
          && fill_pos_ && std::abs(og_pos_ - *fill_pos_) > 1.e-6) {
        *fill_pos_ = og_pos_;
        SPDLOG_INFO(
            "{} Update {} fill_pos with og_pos {}", upd.timestamp(), order_symbol_, og_pos_);
      }
    }
  }

  void onAccountOrder(
      const IProduct& product,
      const OrderUpdate& upd) {
    last_timestamp_ = upd.timestamp();
    if (upd.product_order_info().event().type() == coin::proto::OrderEvent::ORDER_FILLED
        && fill_pos_.has_value()) {
      last_fill_ts_ = upd.timestamp();
      CHECK_GT(last_fill_ts_, 0);
      double fill_qty = upd.product_order_info().event().fill_qty();
      if (upd.product_order_info().event().trade_side() == coin::proto::TRADE_BUY_SIDE) {
        *fill_pos_ += fill_qty;
      } else {
        *fill_pos_ -= fill_qty;
      }

      SPDLOG_INFO(
          "{} {} fill_pos {} {}", last_fill_ts_, order_symbol_, *fill_pos_, og_pos_);
    }
  }

  const std::vector<const IOrderContext*>& GetAllProductWorkingOrders() const {
    return oq_->GetAllProductWorkingOrders(order_symbol_);
  }

  double GetPosition(const IProduct& product) const {
    if (last_fill_ts_ > 0 && fill_pos_) {
      if (::coin2::base::LogEveryDuration(&lastlogt_fillpos_, last_timestamp_, 10'000'000'000L)) {
        SPDLOG_INFO("ret {} fill_pos {}", order_symbol_, *fill_pos_);
      }
      return *fill_pos_;
    } else {
      double ret = oq_->position(product);
      if (::coin2::base::LogEveryDuration(&lastlogt_pos_, last_timestamp_, 10'000'000'000L)) {
        if (!config_.disable_verbose_logging())
          SPDLOG_INFO("ret {} pos {}", order_symbol_, ret);
      }
      return ret;
    }
  }

  double GetRelativePosition(const IProduct& product) const {
    if (config_.qty().has_min_position() && config_.qty().has_max_position()) {
      double pos = GetPosition(product);
      double mean_pos = 0.5 * (
          config_.qty().min_position() + config_.qty().max_position());
      return (pos - mean_pos);
    } else {
      return 0;
    }
  }

  bool HasPosition(const IProduct& product) const {
    return oq_->has_position_info(product);
  }

  // ranges from [-stack, stack] = [-min_pos/lot_size, max_pos/lot_size]
  double GetStack(const IProduct& product) const {
    if (config_.qty().has_min_position() && config_.qty().has_max_position()) {
      double pos = GetPosition(product);
      double mean_pos = 0.5 * (
          config_.qty().min_position() + config_.qty().max_position());
      return (pos - mean_pos) / config_.qty().max_lot_size();
    } else {
      return 0;
    }
  }

  double GetStackAbsolute(const IProduct& product) const {
    double pos = GetPosition(product);
    return pos / config_.qty().max_lot_size();
  }

  OrderQuerier* oq() {
    return oq_;
  }

  int64_t GetSubmitCnt() const {
    return submit_cnt_;
  }
  int64_t GetCancelCnt() const {
    return cancel_cnt_;
  }

 private:
  bool size_shape_configured_ = false;
  std::optional<double> fill_pos_ = std::nullopt;
  int64_t last_fill_ts_{0};
  double og_pos_{0};
  int64_t og_pos_ts_{0};
  bool use_fill_pos_ = true;
  double update_pos_with_og_pos_period_ = 0;
  // og pos consolidation end

 public:
  OrderExecutorConfig* GetMutableOrigConfig() {
    return &config_orig_;
  }

  const OrderExecutorConfig& GetConfig() const {
    return config_;
  }

  void set_order_system(IOrderSubsystem* os) {
    os_ = os;
    if (os->is_sim_os()) {
      oq_ = os_->GenOrderQuerierForSim(config_.sim_account_id());
    } else {
      oq_ = os_->GenOrderQuerier();
    }
    CHECK(oq_);
  }

  /* --------------------------------- periodic hooks --------------------------------- */
  void onPeriodicOrderCheck(int64_t timestamp) {
    if (os_ != nullptr) {
      if (config_.cancel().order_lifetime_ns() != 0) {
        if (timestamp < prev_check_ts_ + 50e6) {
          return;
        }
        prev_check_ts_ = timestamp;

        for (const auto* order :
              oq_->GetAllProductWorkingOrders(order_symbol_)) {
          if (order->order_created_time()
                  + config_.cancel().order_lifetime_ns() < timestamp &&
              order->order_state() == OrderState::WORKING_ORDER) {
            if (CancelOrder(timestamp, order->proc_order_id(), order)) {
              SPDLOG_WARN(
                  "[OrderExecutor] Found an old order created at {} Canceling oid={}",
                  order->order_created_time(),
                  order->proc_order_id());
            }
          }
        }
      }
    }
  }

  int CancelAllOrders(OrderDirection direction, int64_t timestamp) {
    int ret = 0;
    for (const auto* order :
          oq_->GetAllProductWorkingOrders(order_symbol_)) {
      if (order->order_state() == OrderState::WORKING_ORDER &&
          order->direction() == direction) {
        if (CancelOrder(timestamp, order->proc_order_id(), order)) ret++;
      }
    }
    return ret;
  }

  bool CancelOrderAll() {
    ++cancel_cnt_;
    return os_->gateway()->CancelOrderAll(order_symbol_);
  }

  void ForceCancelAllOrders() {
    for (const auto* order : oq_->GetAllProductWorkingOrders(order_symbol_)) {
      if (order->order_state() == OrderState::WORKING_ORDER) {
        os_->gateway()->CancelOrder(order->proc_order_id());
        ++cancel_cnt_;
      }
    }
  }

  bool MakeConversion(
      const coin2::exchange::base::symbology::ICurrency& from_currency,
      const coin2::exchange::base::symbology::ICurrency& to_currency,
      double from_amount) {
    auto conversion_spec = coin2::exchange::base::order::SpecManager::GenerateConversionSpec(
        from_currency,
        to_currency,
        from_amount);
    return os_->gateway()->SubmitConversion(conversion_spec);
  }

  bool MakeConversion(
      const std::string& from_currency,
      const std::string& to_currency,
      double from_amount) {
    auto conversion_spec = coin2::exchange::base::order::SpecManager::GenerateConversionSpec(
        from_currency,
        to_currency,
        from_amount);
    return os_->gateway()->SubmitConversion(conversion_spec);
  }

  const IOrderContext* GetInsideOrder(OrderDirection direction) const {
    const IOrderContext* ret = nullptr;
    for (const auto* order :
          oq_->GetAllProductWorkingOrders(order_symbol_)) {
      CHECK_EQ(order->native_product(), order_symbol_);
      if (direction != order->direction()) {
        continue;
      }
      if (!ret) {
        ret = order;
        continue;
      }
      if (direction == OrderDirection::DIRECTION_BUY) {
        if (ret->order_price() < order->order_price()) {
          ret = order;
        }
      } else if (direction == OrderDirection::DIRECTION_SELL) {
        if (ret->order_price() > order->order_price()) {
          ret = order;
        }
      }
    }
    return ret;
  }

  void SetSimTimestamp(int64_t timestamp) {
    os_->gateway()->UpdateTimeForSim(timestamp);
  }

  bool CancelOrder(
      int64_t timestamp,
      int64_t os_oid,
      const IOrderContext* order_ptr = nullptr) {
    if (order_ptr == nullptr) {
      order_ptr = oq_->FindByProcOrderId(os_oid);
    }
    if (order_ptr == nullptr) {
      return false;
    }
    const auto& order = *order_ptr;
    if (!order.has_exchange_order_id() && os_->gateway()->need_exchange_id_to_cancel()) {
      // when we can't cancel before accept
      return false;
    }

    CHECK_EQ(order.proc_order_id(), os_oid);
    if (cancel_cooldown_limiter_.InCoolDown(timestamp, order.proc_order_id())) {
      if (!config_.disable_verbose_logging())
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "Cancel aborted due to the cancel cooldown: {}",
            order.proc_order_id());
      return false;
    } else {
      if (!config_.disable_verbose_logging())
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "  Canceling oid={}",
            order.proc_order_id());
      cancel_cooldown_limiter_.Update(timestamp, order.proc_order_id());
      os_->gateway()->CancelOrder(order.proc_order_id());
      ++cancel_cnt_;
      return true;
    }
  }

  int64_t LastSubmitTimestamp(OrderDirection direction) const {
    switch (direction) {
      case OrderDirection::DIRECTION_BUY:
        return last_buy_ts_;
      case OrderDirection::DIRECTION_SELL:
        return last_sell_ts_;
      default:
        return 0;
    }
  }

  /* -------------------------------------------------------------------------- */
  /*                                Order Actions                               */
  /* -------------------------------------------------------------------------- */
  // return internal_oid
  OrderExecutionResult SubmitOrder(
      const IProductHolder& product_holder,  // we need product info for ticksize
      double raw_price,                      // adjusted = fitted to tick
      double raw_qty,
      OrderDirection direction,         // mapped to BUY_OPEN, BUY_CLOSE, SELL_OPEN, SELL_CLOSE
      OrderDuration duration,           //
      OrderType order_type,             // MARKET, STOP, LIMIT, STOP_LIMIT
      std::optional<double> bbo_price,  // used for safety
      double leverage_rate,             // defaults to config.leverage
      bool post_only,
      const std::string& tag,
      int64_t timestamp,
      bool log_fail,
      bool block_price_band,
      std::optional<double> maxpos_multiplier,
      std::optional<bool> reduce_only,
      bool force_submit) {
    /* ------------------------------ params check ----------------------------- */
    CHECK_NE_THROW(direction, OrderDirection::DIRECTION_UNINITIALIZED);
    CHECK_NE_THROW(direction, OrderDirection::DIRECTION_SPECIAL)
        << "special OrderDirection is not implemented yet";
    if (bbo_price) {
      CHECK_GT_THROW(*bbo_price, 0);
    }
    CHECK_GT_THROW(leverage_rate, 0);

    /* ------------------------------ product check ----------------------------- */
    const auto& product = product_holder.product();

    if (!os_->is_sim_os() &&
        config_.block_maker_maxrate_use_ratio() < 1 &&
        os_->gateway()->GetMutableRateTracker()) {
      double used_ratio = os_->gateway()->GetMutableRateTracker()->GetMaxRateUsedRatio(
            timestamp, ActionType::PLACE);
      SPDLOG_INFO_EVERY_NS(
          5'000'000'000L,
          timestamp,
          "Rate used up ratio: {}", used_ratio);
      if (post_only && used_ratio > config_.block_maker_maxrate_use_ratio()) {
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "{}: Rate used up too much, block post only submission",
            product.symbol());
        return {};
      }
    }

    if (!oq_->has_position_info(product)) {
      SPDLOG_WARN("{} position not ready yet, skipping.", product.symbol());
      return {};
    }

    /* -------------------------------- qty check ------------------------------- */
    CHECK_GT_THROW(raw_qty, 0) << product.symbol() << ", price: " << raw_price;

    // raw_qty => min(round(raw_qty), pos, lot_size)
    const auto& tick_qty_policy = product_holder.product_info().tick_qty();
    double qty = tick_qty_policy.RoundWithPolicy(raw_qty, direction, config_.qty().round());

    qty = CheckMaxLotSize(qty, tick_qty_policy);

    bool china_hard_close = false;

    if (config_.side().is_china_style() &&
        config_.qty().min_position() == 0 &&
        config_.qty().max_position() == 0 &&
        std::abs(GetPosition(product)) < product_holder.product_info().tick_qty().submittable_min_qty() &&
        oq_->open_long_avail_qty(product) > 0 &&
        oq_->open_short_avail_qty(product) > 0) {
      // clear mode, let's buy as long as BUY_CLOSE is available
      // sell: vice versa
      // thus don't check max position here, naturally it will pass through to GetChinaStyleQty().
      china_hard_close = true;
      china_soft_closed_ = true;
    } else {
      if (direction == OrderDirection::DIRECTION_BUY) {
        if (log_fail) {
          SPDLOG_INFO("[Executor - SubmitOrder] Buy Qty Check");
        }
        qty = CheckMaxPositionForBuy(
            product,
            qty,
            tick_qty_policy,
            0,  // pending_qty
            maxpos_multiplier);
      } else if (direction == OrderDirection::DIRECTION_SELL) {
        if (log_fail) {
          SPDLOG_INFO("[Executor - SubmitOrder] Sell Qty Check");
        }
        qty = CheckMinPositionForSell(
            product,
            qty,
            tick_qty_policy,
            0,  // pending_qty,
            maxpos_multiplier);
      }
    }

    if (qty == 0) {
      if (log_fail) {
        if (!config_.disable_verbose_logging())
          SPDLOG_INFO("[Executor - SubmitOrder] Qty Fail!");
      }
      return {};
    } else {
      if (!config_.disable_verbose_logging())
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "[Executor - SubmitOrder] Qty {}", qty);
    }

    /* ------------------------------- price check ------------------------------ */
    CHECK_GT_THROW(raw_price, 0);

    // raw price => adjusted price
    const auto& tick_price_policy = product_holder.product_info().tick_price();
    double price = tick_price_policy.RoundWithPolicy(raw_price, direction, config_.price().round());

    int64_t diff_ns = timestamp - LastSubmitTimestamp(direction);
    // order throttle
    if (diff_ns < config_.order_update_period_ns()) {
      if (log_fail) {
        SPDLOG_INFO("[Executor - SubmitOrder] order update period Fail! {}", diff_ns);
      }
      return {};
    } else if (block_price_band) {
      NOTREACHED() << "block_price_band is decommed";
    }

    /* -------------------------- min_amount (pq) check ------------------------- */
    // some exchange venues (e.g., Binance USDm Futures) have min_amount (pq) requirement
    // so if qty * price is smaller than the required min_amount, skip submit
    // only for not reduce only within futures or within spot.
    if (!isnan(product_holder.product_info().min_amount()) &&
        !(product.market() == MarketType::Futures &&
            reduce_only.has_value() && reduce_only.value())) {
      if (qty * price < product_holder.product_info().min_amount()) {
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "[Executor - SubmitOrder] skip submit due to Qty {} * Price {} < min_amount {}",
            qty, price, product_holder.product_info().min_amount());
        return {};
      }
    }

    /* ------------------------------- order side ------------------------------- */
    // This behavior differs for Spot/Futures

    CHECK_GE_THROW(order_type, 1);
    std::vector<SideQty> side_qtys;

    if (product.market() == MarketType::Spot || (
            product.market() == MarketType::Futures &&
            product.exchange() == ExchangeType::Bitflyer &&
            StringEndsWith(product.relative_norm(), "IMMEDIATE"))) {
      qty = CheckMinOrderQty(qty, force_submit);
      if (qty > 0) {
        if (direction == OrderDirection::DIRECTION_BUY) {
          side_qtys.push_back({OrderSide::BUY_ORDER, qty});  // qty is already rounded
        } else if (direction == OrderDirection::DIRECTION_SELL) {
          side_qtys.push_back({OrderSide::SELL_ORDER, qty});
        } else {
          throw std::runtime_error("Unknown OrderDirection");
        }
      }
    } else if (product.market() == MarketType::Futures) {
      if (config_.side().is_china_style()) {
        GetChinaStyleQty(
            product,
            direction,
            CheckMinOrderQty(qty, force_submit),
            config_.side().china_style_policy(),
            (china_soft_closed_)? 0.0 : config_.side().maintain_qty(),
            china_hard_close,
            &side_qtys);
      } else {
        if (direction == OrderDirection::DIRECTION_BUY) {
          qty = CheckMinOrderQty(qty, force_submit);
          if (qty > 0) {
            side_qtys.push_back({OrderSide::BUY_ORDER, qty});
          }

        } else {
          qty = CheckMinOrderQty(qty, force_submit);
          if (qty > 0) {
            side_qtys.push_back({OrderSide::SELL_ORDER, qty});
          }
        }
      }
    } else {
      throw std::runtime_error("[OrderExecutor] MarketType is other than [Spot, Futures]");
    }
    if (config_.qty().split_lot_size() > 0
        && side_qtys.size() == 1
        && side_qtys[0].qty > config_.qty().split_lot_size()) {
      double remain_qty = side_qtys[0].qty - config_.qty().split_lot_size();
      side_qtys[0].qty = config_.qty().split_lot_size();
      if (remain_qty > config_.qty().split_lot_size()) {
        LOG(ERROR) << "Qty exceeds 2 * split_size. please consider downsizing";
      }
      remain_qty = std::min(remain_qty, config_.qty().split_lot_size());
      side_qtys.push_back({side_qtys[0].side, remain_qty});
    }

    /* ------------------------------ submit orders ----------------------------- */
    OrderExecutionResult ret;

    for (auto side_qty : side_qtys) {
      // oid is generated by OG
      OrderSpec order_spec(
          product,
          order_type,
          side_qty.side,
          duration,
          price,
          side_qty.qty,
          timestamp,  // order_created_time
          post_only,
          leverage_rate,
          tag);
      if (reduce_only.has_value() && reduce_only.value()) {
        order_spec.SetReduceOnly(true);
      }
      // succesfully submitted
      if (direction == OrderDirection::DIRECTION_BUY) {
        prev_buy_price_ = price;
        last_buy_ts_ = timestamp;
      } else if (direction == OrderDirection::DIRECTION_SELL) {
        prev_sell_price_ = price;
        last_sell_ts_ = timestamp;
      }

      // dryrun? log but not fire
      if (config_.dryrun()) {
        continue;
      }
      std::optional<int64_t> oid;
      if (config_.has_sim_account_id()) {
        oid = os_->gateway()->SubmitOrderForSim(
            config_.sim_account_id(), order_spec);
        ++submit_cnt_;
      } else {
        oid = os_->gateway()->SubmitOrder(order_spec);
        ++submit_cnt_;
      }
      if (!oid) {
        continue;
      }

      ret.qty += side_qty.qty;
      if (ret.first_oid != 0) {
        ret.second_oid = *oid;
      } else {
        ret.first_oid = *oid;
      }

      if (!config_.disable_verbose_logging())
        SPDLOG_INFO("mea {} order_spec {}", os_->mea().String(), order_spec.DebugString());
    }
    return ret;
  }

  // This method is introduced for Prex.
  // If you are using AmendOrder for another exchange, make sure it is working properly.
  // China Style is unsupported.
  OrderExecutionResult AmendOrder(
      const IProductHolder& product_holder,  // we need product info for ticksize
      int64_t orig_proc_order_id,
      double raw_price,                      // adjusted = fitted to tick
      double raw_qty,
      OrderDirection direction,         // mapped to BUY_OPEN, BUY_CLOSE, SELL_OPEN, SELL_CLOSE
      bool post_only,
      int64_t timestamp,
      bool log_fail,
      std::optional<double> maxpos_multiplier,
      std::optional<bool> reduce_only,
      bool force_submit,
      double orig_order_qty = 0.0) {
    /* ------------------------------ params check ----------------------------- */
    CHECK_NE_THROW(direction, OrderDirection::DIRECTION_UNINITIALIZED);
    CHECK_NE_THROW(direction, OrderDirection::DIRECTION_SPECIAL)
        << "special OrderDirection is not implemented yet";

    /* ------------------------------ product check ----------------------------- */
    const auto& product = product_holder.product();

    if (!os_->is_sim_os() &&
        config_.block_maker_maxrate_use_ratio() < 1 &&
        os_->gateway()->GetMutableRateTracker()) {
      double used_ratio = os_->gateway()->GetMutableRateTracker()->GetMaxRateUsedRatio(
            timestamp, ActionType::PLACE);
      SPDLOG_INFO_EVERY_NS(
          5'000'000'000L,
          timestamp,
          "Rate used up ratio: {}", used_ratio);
      if (post_only && used_ratio > config_.block_maker_maxrate_use_ratio()) {
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "{}: Rate used up too much, block post only submission",
            product.symbol());
        return {};
      }
    }

    if (!oq_->has_position_info(product)) {
      SPDLOG_WARN("{} position not ready yet, skipping.", product.symbol());
      return {};
    }

    /* -------------------------------- qty check ------------------------------- */
    CHECK_GT_THROW(raw_qty, 0) << product.symbol() << ", price: " << raw_price;

    // raw_qty => min(round(raw_qty), pos, lot_size)
    const auto& tick_qty_policy = product_holder.product_info().tick_qty();
    double qty = tick_qty_policy.RoundWithPolicy(raw_qty, direction, config_.qty().round());

    qty = CheckMaxLotSize(qty, tick_qty_policy);

    if (direction == OrderDirection::DIRECTION_BUY) {
      if (log_fail) {
        SPDLOG_INFO("[Executor - SubmitOrder] Buy Qty Check");
      }
      qty = CheckMaxPositionForBuy(
          product,
          qty,
          tick_qty_policy,
          0,  // pending_qty
          maxpos_multiplier,
          orig_order_qty);
    } else if (direction == OrderDirection::DIRECTION_SELL) {
      if (log_fail) {
        SPDLOG_INFO("[Executor - SubmitOrder] Sell Qty Check");
      }
      qty = CheckMinPositionForSell(
          product,
          qty,
          tick_qty_policy,
          0,  // pending_qty
          maxpos_multiplier,
          orig_order_qty);
    }

    if (qty == 0) {
      if (log_fail) {
        if (!config_.disable_verbose_logging())
          SPDLOG_INFO("[Executor - SubmitOrder] Qty Fail!");
      }
      return {};
    } else {
      if (!config_.disable_verbose_logging())
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "[Executor - SubmitOrder] Qty {}", qty);
    }

    /* ------------------------------- price check ------------------------------ */
    CHECK_GT_THROW(raw_price, 0);

    // raw price => adjusted price
    const auto& tick_price_policy = product_holder.product_info().tick_price();
    double price = tick_price_policy.RoundWithPolicy(raw_price, direction, config_.price().round());

    int64_t diff_ns = timestamp - LastSubmitTimestamp(direction);
    // order throttle
    if (diff_ns < config_.order_update_period_ns()) {
      if (log_fail) {
        SPDLOG_INFO("[Executor - SubmitOrder] order update period Fail! {}", diff_ns);
      }
      return {};
    }

    /* -------------------------- min_amount (pq) check ------------------------- */
    // some exchange venues (e.g., Binance USDm Futures) have min_amount (pq) requirement
    // so if qty * price is smaller than the required min_amount, skip submit
    // only for not reduce only within futures or within spot.
    if (!isnan(product_holder.product_info().min_amount()) &&
        !(product.market() == MarketType::Futures &&
            reduce_only.has_value() && reduce_only.value())) {
      if (qty * price < product_holder.product_info().min_amount()) {
        SPDLOG_INFO_EVERY_NS(
            5'000'000'000L,
            timestamp,
            "[Executor - SubmitOrder] skip submit due to Qty {} * Price {} < min_amount {}",
            qty, price, product_holder.product_info().min_amount());
        return {};
      }
    }

    /* ------------------------------- order side ------------------------------- */
    // This behavior differs for Spot/Futures

    std::vector<SideQty> side_qtys;

    if (product.market() == MarketType::Spot) {
      qty = CheckMinOrderQty(qty, force_submit);
      if (qty > 0) {
        if (direction == OrderDirection::DIRECTION_BUY) {
          side_qtys.push_back({OrderSide::BUY_ORDER, qty});  // qty is already rounded
        } else if (direction == OrderDirection::DIRECTION_SELL) {
          side_qtys.push_back({OrderSide::SELL_ORDER, qty});
        } else {
          throw std::runtime_error("Unknown OrderDirection");
        }
      }
    } else if (product.market() == MarketType::Futures) {
      if (direction == OrderDirection::DIRECTION_BUY) {
        qty = CheckMinOrderQty(qty, force_submit);
        if (qty > 0) {
          side_qtys.push_back({OrderSide::BUY_ORDER, qty});
        }

      } else {
        qty = CheckMinOrderQty(qty, force_submit);
        if (qty > 0) {
          side_qtys.push_back({OrderSide::SELL_ORDER, qty});
        }
      }
    } else {
      throw std::runtime_error("[OrderExecutor] MarketType is other than [Spot, Futures]");
    }

    /* ------------------------------ submit orders ----------------------------- */
    OrderExecutionResult ret;

    for (auto side_qty : side_qtys) {
      // oid is generated by OG
      // succesfully submitted
      if (direction == OrderDirection::DIRECTION_BUY) {
        prev_buy_price_ = price;
        last_buy_ts_ = timestamp;
      } else if (direction == OrderDirection::DIRECTION_SELL) {
        prev_sell_price_ = price;
        last_sell_ts_ = timestamp;
      }

      // dryrun? log but not fire
      if (config_.dryrun()) {
        continue;
      }
      std::optional<int64_t> oid;
      if (config_.has_sim_account_id()) {
        // TODO(hgpark, peter) : implement AmendOrderForSim
        /*
        oid = os_->gateway()->SubmitOrderForSim(
            config_.sim_account_id(), orig_proc_order_id, price, side_qty.qty);
        */
        ++submit_cnt_;
      } else {
        oid = os_->gateway()->AmendOrder(orig_proc_order_id, price, side_qty.qty);
        ++submit_cnt_;
      }
      if (!oid) {
        continue;
      }

      ret.qty += side_qty.qty;
      if (ret.first_oid != 0) {
        ret.second_oid = *oid;
      } else {
        ret.first_oid = *oid;
      }

      if (!config_.disable_verbose_logging())
        SPDLOG_INFO("mea {} orig_proc_order_id {} price {} qty {}",
                    os_->mea().String(), orig_proc_order_id, price, side_qty.qty);
    }
    return ret;
  }

  CoolDownLimiter* GetMutableCancelCoolDownLimiter() {
    return &cancel_cooldown_limiter_;
  }

 public:
  // layering impl start
  double GetMaximumWorkingQty(const IProduct& product, int sign) {
    if (sign > 0) {
      return config_.qty().max_position() - GetPosition(product);
    } else {
      return GetPosition(product) - config_.qty().min_position();
    }
  }

  std::vector<const IOrderContext*> GetWorkingOrders(int sign) {
    OrderDirection direction = (sign > 0) ?
        OrderDirection::DIRECTION_BUY : OrderDirection::DIRECTION_SELL;
    std::vector<const IOrderContext*> rows;
    for (const auto* order : oq_->GetAllProductWorkingOrders(order_symbol_)) {
      if (order->order_state() == OrderState::WORKING_ORDER && order->direction() == direction) {
        rows.push_back(order);
      }
    }
    return rows;
  }

  void ManageSideOrders(
      const IProductHolder& product_holder,
      int64_t timestamp,
      int sign,
      double leverage_rate,
      LayeringLevels levels) {
    // levels.ClipQty(GetMaximumWorkingQty(product_holder.product(), sign));

    CHECK_EQ(levels.sign, sign);
    auto order_list = GetWorkingOrders(sign);
    std::sort(order_list.begin(), order_list.end(), [sign](const auto* wo1, const auto* wo2) {
      if (sign < 0) {
        return wo1->order_price() < wo2->order_price();
      } else {
        return wo1->order_price() > wo2->order_price();
      }
      });
    auto opers = LayeringLevels::GetMostDesiredModification(levels, timestamp, sign, order_list);
    for (const auto& oper : opers) {
      if (oper.opertype != LayeringLevels::NOOP) {
        LOG(INFO) << oper.opertype << ") sgn:" << sign << " " << oper.qty << " @ " << oper.price;
        ExecuteOperation(product_holder, timestamp, sign, leverage_rate, oper);
      }
    }
  }

  void ExecuteOperation(
      const IProductHolder& product_holder,
      int64_t timestamp,
      int sign,
      double leverage_rate,
      const LayeringLevels::Level& oper) {
    OrderDirection direction = (sign > 0) ?
        OrderDirection::DIRECTION_BUY : OrderDirection::DIRECTION_SELL;
    switch (oper.opertype) {
      case LayeringLevels::CANCEL:
        os_->gateway()->CancelOrder(*oper.proc_order_id);
        break;
      case LayeringLevels::POST:
        SubmitOrder(
            product_holder,
            oper.price,
            oper.qty,
            direction,
            OrderDuration::GTC_ORDER,
            OrderType::LIMIT_ORDER,
            std::nullopt,               // optional<double> bbo_price
            leverage_rate,              // defaults to config.leverage
            true,
            "maker_layer",
            timestamp,
            false,
            false,
            std::nullopt,
            std::nullopt,
            false);
        break;
      default:
        NOTREACHED() << oper.opertype;
    }
  }
  // layering end

 protected:
  /* --------------------------------- helpers -------------------------------- */

  constexpr static double BP = 1e-4;
  bool FindOrderWithinBp(OrderDirection direction, double price, double bp) {
    auto orders = oq_->GetAllProductWorkingOrders(order_symbol_);
    for (auto* order : orders) {
      if (order->direction() == direction &&
          std::abs(order->order_price() - price) / (order->order_price() + price) * 2 <=
              bp * BP) {
        return true;
      }
    }
    return false;
  }

  bool FindOrderWithPrice(OrderDirection direction, double price) {
    auto orders = oq_->GetAllProductWorkingOrders(order_symbol_);
    for (auto* order : orders) {
      if (order->direction() == direction && order->order_price() == price) {
        return true;
      }
    }
    return false;
  }

  /* ------------------------------- qty checks ------------------------------- */
  double CheckMinOrderQty(double qty, bool force_submit = false) {
    if (config_.qty().has_min_order_qty()) {
      if (qty < config_.qty().min_order_qty()) {
        if (qty > 0 && force_submit)
          return config_.qty().min_order_qty();
        return 0;
      }
    }
    return qty;
  }

  double CheckMaxLotSize(
      double qty,
      const coin2::exchange::base::symbology::TickQty& policy,
      double pending_qty = 0) {
    if (config_.qty().has_max_lot_size()) {
      double available = config_.qty().max_lot_size() - pending_qty;
      if (qty > available) {
        if (available < 0) {
          return 0;
        }
        return policy.Floor(available);
      }
    }
    return qty;
  }

  double CheckMaxPositionForBuy(
      const IProduct& product,
      double qty,
      const coin2::exchange::base::symbology::TickQty& policy,
      double pending_qty = 0,
      std::optional<double> maxpos_multiplier = std::nullopt,
      double orig_order_qty = 0.0) {
    // max net position
    if (config_.qty().has_max_position()) {
      double max_pos = config_.qty().max_position();
      if (maxpos_multiplier) {
        max_pos *= *maxpos_multiplier;
      }
      double buy_available =
          max_pos - (
              GetPosition(product)
              // use_fill_pos == false : get qty by potential fill pos == true
              + oq_->GetBuyWorkingOrderQty(product, !use_fill_pos_)
              + pending_qty) + orig_order_qty;
      if (qty > buy_available) {
        if (buy_available < 0) {
          return 0;
        }
        return policy.Floor(buy_available);
      }
    }
    return qty;
  }

  double CheckMinPositionForSell(
      const IProduct& product,
      double qty,
      const TickQty& policy,
      double pending_qty = 0,
      std::optional<double> maxpos_multiplier = std::nullopt,
      double orig_order_qty = 0.0) {
    // min net position
    if (config_.qty().has_min_position()) {
      // e.g. min_position = -10
      //      position = -4
      //      GetSellWorkingOrderQty = 5
      //      -min +pos-sell = 1
      double min_pos = config_.qty().min_position();
      if (maxpos_multiplier) {
        min_pos *= *maxpos_multiplier;
      }
      double sell_available = (
          GetPosition(product)
              // use_fill_pos == false : get qty by potential fill pos == true
          - oq_->GetSellWorkingOrderQty(product, !use_fill_pos_)
          - pending_qty) - min_pos + orig_order_qty;
      if (qty > sell_available) {
        if (sell_available < 0) {
          return 0;
        }
        return policy.Floor(sell_available);
      }
    }
    return qty;
  }

  void GetChinaStyleQty(
      const IProduct& product,
      OrderDirection direction,
      double qty,
      ChinaStyleOrderSidePolicy policy,
      double maintain_qty,
      bool china_hard_close,
      std::vector<SideQty>* side_qtys) const {
    if (china_hard_close) {
      if (direction == OrderDirection::DIRECTION_BUY) {
        double close_available = CheckBuyCloseQty(product, qty);
        if (close_available > 0) {
          side_qtys->push_back({OrderSide::BUY_CLOSE_ORDER, close_available});
        }
      } else if (direction == OrderDirection::DIRECTION_SELL) {
        double close_available = CheckSellCloseQty(product, qty);
        if (close_available > 0) {
          side_qtys->push_back({OrderSide::SELL_CLOSE_ORDER, close_available});
        }
      }
    } else {
      switch (policy) {
        case ChinaStyleOrderSidePolicy::CHINA_POLICY_UNINITIALIZED:
          THROW() << "You should provide ChinaStyleOrderSidePolicy";
          break;
        case ChinaStyleOrderSidePolicy::CHINA_POLICY_MAINTAIN_QTY:
          if (qty > 0) {
            if (direction == OrderDirection::DIRECTION_BUY) {
              maintain_qty = std::max(maintain_qty, qty);
              double close_available = CheckBuyCloseQty(product, maintain_qty);
              if (close_available >= maintain_qty) {
                side_qtys->push_back({OrderSide::BUY_CLOSE_ORDER, qty});
              } else {
                side_qtys->push_back({OrderSide::BUY_OPEN_ORDER, qty});
              }
            } else if (direction == OrderDirection::DIRECTION_SELL) {
              maintain_qty = std::max(maintain_qty, qty);
              double close_available = CheckSellCloseQty(product, maintain_qty);
              if (close_available >= maintain_qty) {
                side_qtys->push_back({OrderSide::SELL_CLOSE_ORDER, qty});
              } else {
                side_qtys->push_back({OrderSide::SELL_OPEN_ORDER, qty});
              }
            } else {
              THROW();
            }
          }
          break;
        default:
          NOTREACHED() << ChinaStyleOrderSidePolicy_Name(policy);
          break;
      }
    }
  }
  double CheckSellCloseQty(const IProduct& product, double qty) const {
    double available =
        oq_->open_long_avail_qty(product) - oq_->GetSellCloseWorkingOrderQty(product);
    return std::max(0.0, std::min(qty, available));
  }
  double CheckBuyCloseQty(const IProduct& product, double qty) const {
    double available =
        oq_->open_short_avail_qty(product) - oq_->GetBuyCloseWorkingOrderQty(product);
    return std::max(0.0, std::min(qty, available));
  }

 private:
  const std::string name_;
  const std::string order_symbol_;
  OrderExecutorConfig config_orig_;
  OrderExecutorConfig config_;
  IOrderSubsystem* os_;
  OrderQuerier* oq_;

  int64_t last_buy_ts_;
  int64_t last_sell_ts_;
  int64_t prev_check_ts_;
  int64_t submit_cnt_;
  int64_t cancel_cnt_;
  CoolDownLimiter cancel_cooldown_limiter_ = CoolDownLimiter(1'500'000'000L);

  std::optional<double> prev_buy_price_;
  std::optional<double> prev_sell_price_;

  bool china_soft_closed_ = false;
};

// 1 OrderExecutor handles 1 MEA
class OrderExecutorSystem {
  // using IOrderSubsystem = coin2:;exchange::base::order::

 public:
  ////////// Methods for Strategy //////////
  OrderExecutorSystem(
      const std::string& name,
      const MarketExchangeApi& mea,
      const exchange::di::ConstantInjection& ci,
      const OrderExecutorSystemConfig& config,
      int64_t timestamp,
      IOrderSubsystem* os = nullptr)
      : name_(name), mea_(mea), system_config_(config), os_(os) {
    // File - gz - topic - proto
    // logger_ = std::make_unique<orio::io::ProtoWriter>();

    // create OE for each product
    CHECK_GT(timestamp, 0);
    coin2::exchange::symbology::ProductRequest req;
    req.set_mea(mea.String());
    for (const auto& [prod, product_group] : config.products()) {
      req.set_norm(prod);
      for (auto&& product:
          ResolveProductRequestFromPotentiallyWildcard(req, timestamp, false)) {
        auto symbol = product->absolute_norm();
        if (executors_.find(symbol) != executors_.end()) {
          throw std::runtime_error("[OrderExecutorSystem] duplicated symbol: " + symbol);
        }
        executors_[symbol] = std::make_unique<OrderExecutor>(
            mea.exchange,
            name,
            product->order_native_symbol(),
            product_group,
            os);
      }
    }
  }

  /* -------------------------------------------------------------------------- */
  /*                               Getter / Setter                              */
  /* -------------------------------------------------------------------------- */
  const std::string& name() const { return name_; }
  const OrderExecutorSystemConfig& config() const { return system_config_; }
  const MarketExchangeApi& mea() const { return mea_; }
  IOrderSubsystem* order_system() { return os_; }
  OrderQuerier* order_querier() { return oq_; }
  void set_order_system(IOrderSubsystem* os) {
    std::optional<int64_t> account_id;
    for (const auto& [_, prod_config] : system_config_.products()) {
      if (account_id) {
        CHECK_EQ(*account_id, prod_config.sim_account_id());
      } else {
        account_id = prod_config.sim_account_id();
      }
    }
    os_ = os;
    if (os->is_sim_os()) {
      oq_ = os_->GenOrderQuerierForSim(account_id ? *account_id : 0L);
    } else {
      oq_ = os_->GenOrderQuerier();
    }
    for (auto& [symbol, oe] : executors_) {
      (void)symbol;
      oe->set_order_system(os);
    }
  }

  OrderQuerier* oq(const IProductHolder& product_holder) {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->oq();
  }

  bool HasProduct(const IProductHolder& product_holder) const {
    return executors_.count(product_holder.product().absolute_norm()) > 0;
  }

  bool HasPosition(const IProductHolder& product_holder) const {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->HasPosition(product_holder.product());
  }

  double GetCurrentStack(const IProductHolder& product_holder) const {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->GetStack(product_holder.product());
  }

  double GetCurrentStackAbsolute(const IProductHolder& product_holder) const {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->GetStackAbsolute(product_holder.product());
  }

  double GetCurrentPosition(const IProductHolder& product_holder) const {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->GetPosition(product_holder.product());
  }

  double GetRelativePosition(const IProductHolder& product_holder) const {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->GetRelativePosition(product_holder.product());
  }

  void onPeriodicOrderCheck(
      const IProductHolder& product_holder,
      int64_t timestamp) {
    auto& oe = executors_.at(product_holder.product().absolute_norm());
    oe->onPeriodicOrderCheck(timestamp);
  }

  bool CancelOrder(
      const IProductHolder& product_holder,
      int64_t timestamp,
      int64_t os_oid) {
    auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->CancelOrder(timestamp, os_oid);
  }

  void AdjustSize(
      const IProductHolder& product_holder,
      double min_position,
      double max_position,
      double max_lot_size,
      std::optional<double> min_order_qty = std::nullopt,
      std::optional<double> maintain_qty = std::nullopt) {
    auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->AdjustSize(
        min_position,
        max_position,
        max_lot_size,
        min_order_qty,
        maintain_qty);
  }

  void SetSimTimestamp(
      const IProductHolder& product_holder,
      int64_t timestamp) {
    auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->SetSimTimestamp(timestamp);
  }

  /* -------------------------------------------------------------------------- */
  /*                                Order Actions                               */
  /* -------------------------------------------------------------------------- */
  // return internal_oid
  OrderExecutionResult SubmitOrder(
      const IProductHolder& product_holder,  // we need product info for ticksize
      double raw_price,                      // adjusted = fitted to tick
      double raw_qty,
      OrderDirection direction,         // mapped to BUY_OPEN, BUY_CLOSE, SELL_OPEN, SELL_CLOSE
      OrderDuration duration,           //
      OrderType order_type,             // MARKET, STOP, LIMIT, STOP_LIMIT
      std::optional<double> bbo_price,  // used for safety
      double leverage_rate,             // defaults to config.leverage
      bool post_only,
      const std::string& tag,
      int64_t timestamp,
      bool log_fail = true,
      bool block_price_band = false,
      std::optional<double> maxpos_multiplier = std::nullopt,
      std::optional<bool> reduce_only = std::nullopt,
      bool force_submit = false) {
    CHECK_GE(timestamp, 0);
    /* --------------------------------- logging -------------------------------- */
    OrderExecutorSubmitOrderLog log;
    log.set_absolute_norm(product_holder.product().absolute_norm());
    log.set_raw_price(raw_price);
    log.set_raw_qty(raw_qty);
    log.set_direction(direction);
    log.set_duration(duration);
    log.set_order_type(order_type);
    if (bbo_price) {
      log.set_bbo_price(*bbo_price);
    }
    log.set_leverage_rate(leverage_rate);
    log.set_post_only(post_only);
    log.set_tag(tag);

    /* ------------------------------ product check ----------------------------- */
    const auto& product = product_holder.product();
    if (product.exchange() != ExchangeType::Bitflyer) {
      CHECK_EQ_THROW(product.market(), mea_.market);
    }
    CHECK_EQ_THROW(product.exchange(), mea_.exchange);

    {
      auto sym = product_holder.product().absolute_norm();
      if (executors_.count(sym) == 0) {
        SPDLOG_INFO("================");
        SPDLOG_INFO("submit order for sym: {}", sym);
        SPDLOG_INFO("available execs: ");
        for (auto& [a, b] : executors_) {
          (void)b;
          SPDLOG_INFO(a);
        }
      }
    }

    auto& oe = executors_.at(product_holder.product().absolute_norm());
    auto result = oe->SubmitOrder(
        product_holder,
        raw_price,
        raw_qty,
        direction,
        duration,
        order_type,
        bbo_price,
        leverage_rate,
        post_only,
        tag,
        timestamp,
        log_fail,
        block_price_band,
        maxpos_multiplier,
        reduce_only,
        force_submit);

    if (result.first_oid != 0) {
      log.add_os_order_ids(result.first_oid);

      if (result.second_oid != 0) {
        log.add_os_order_ids(result.second_oid);
      }
    }

    if (log_fail || result.first_oid != 0) {
      if (!system_config_.disable_verbose_logging())
        SPDLOG_INFO(log.ShortDebugString());
    }

    return result;
  }

  OrderExecutionResult AmendOrder(
      const IProductHolder& product_holder,  // we need product info for ticksize
      int64_t orig_proc_order_id,
      double raw_price,                      // adjusted = fitted to tick
      double raw_qty,
      OrderDirection direction,         // mapped to BUY_OPEN, BUY_CLOSE, SELL_OPEN, SELL_CLOSE
      bool post_only,
      int64_t timestamp,
      bool log_fail = true,
      std::optional<double> maxpos_multiplier = std::nullopt,
      std::optional<bool> reduce_only = std::nullopt,
      bool force_submit = false,
      double orig_order_qty = 0.0) {
    CHECK_GE(timestamp, 0);
    /* --------------------------------- logging -------------------------------- */
    OrderExecutorSubmitOrderLog log;
    log.set_absolute_norm(product_holder.product().absolute_norm());
    log.set_raw_price(raw_price);
    log.set_raw_qty(raw_qty);
    log.set_direction(direction);
    log.set_post_only(post_only);

    /* ------------------------------ product check ----------------------------- */
    const auto& product = product_holder.product();
    if (product.exchange() != ExchangeType::Bitflyer) {
      CHECK_EQ_THROW(product.market(), mea_.market);
    }
    CHECK_EQ_THROW(product.exchange(), mea_.exchange);

    {
      auto sym = product_holder.product().absolute_norm();
      if (executors_.count(sym) == 0) {
        SPDLOG_INFO("================");
        SPDLOG_INFO("submit order for sym: {}", sym);
        SPDLOG_INFO("available execs: ");
        for (auto& [a, b] : executors_) {
          (void)b;
          SPDLOG_INFO(a);
        }
      }
    }

    auto& oe = executors_.at(product_holder.product().absolute_norm());
    auto result = oe->AmendOrder(
        product_holder,
        orig_proc_order_id,
        raw_price,
        raw_qty,
        direction,
        post_only,
        timestamp,
        log_fail,
        maxpos_multiplier,
        reduce_only,
        force_submit,
        orig_order_qty);

    if (result.first_oid != 0) {
      log.add_os_order_ids(result.first_oid);

      if (result.second_oid != 0) {
        log.add_os_order_ids(result.second_oid);
      }
    }

    if (log_fail || result.first_oid != 0) {
      if (!system_config_.disable_verbose_logging())
        SPDLOG_INFO(log.ShortDebugString());
    }

    return result;
  }

  std::string DebugString(int64_t timestamp) const {
    return oq_->order_manager().DebugString(timestamp);
  }

  int CancelAllOrders(
      const IProductHolder& product_holder,
      OrderDirection direction,
      int64_t timestamp) {
    auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->CancelAllOrders(direction, timestamp);
  }

  bool CancelOrderAll(const IProductHolder& product_holder) {
    auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->CancelOrderAll();
  }

  const OrderExecutorConfig& GetConfig(const IProductHolder& product_holder) const {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->GetConfig();
  }

  bool MakeConversion(
      const IProductHolder& product_holder,
      const coin2::exchange::base::symbology::ICurrency& from_currency,
      const coin2::exchange::base::symbology::ICurrency& to_currency,
      double from_amount) {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->MakeConversion(from_currency, to_currency, from_amount);
  }

  bool MakeConversion(
      const IProductHolder& product_holder,
      const std::string& from_currency,
      const std::string& to_currency,
      double from_amount) {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->MakeConversion(from_currency, to_currency, from_amount);
  }

  OrderExecutorConfig* GetMutableOrigConfig(const IProductHolder& product_holder) {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->GetMutableOrigConfig();
  }

  const IOrderContext* GetInsideOrder(
      const IProductHolder& product_holder, OrderDirection direction) const {
    const auto& oe = executors_.at(product_holder.product().absolute_norm());
    return oe->GetInsideOrder(direction);
  }

  const std::vector<const IOrderContext*>&
    GetAllProductWorkingOrders(const IProductHolder& product_holder) const {
    return executors_.at(
        product_holder.product().absolute_norm())->GetAllProductWorkingOrders();
  }

  void ForceCancelAllWorkingOrders() {
    for (auto&& [name, oe] : executors_) {
      (void)name;
      oe->ForceCancelAllOrders();
    }
  }

  void onAccountInfo(
      const IProductHolder& product_holder,
      const coin2::exchange::base::order::OrderUpdate& upd) {
    executors_.at(product_holder.product().absolute_norm())
        ->onAccountInfo(product_holder.product(), upd);
  }

  void onAccountOrder(
      const IProductHolder& product_holder,
      const coin2::exchange::base::order::OrderUpdate& upd) {
    executors_.at(product_holder.product().absolute_norm())
        ->onAccountOrder(product_holder.product(), upd);
  }

  int64_t GetSubmitCnt() {
    int64_t ret = 0;
    for (auto&& [name, oe] : executors_) {
      ret += oe->GetSubmitCnt();
    }
    return ret;
  }
  int64_t GetCancelCnt() {
    int64_t ret = 0;
    for (auto&& [name, oe] : executors_) {
      ret += oe->GetCancelCnt();
    }
    return ret;
  }

  auto* GetMutableExecutorSub(const IProductHolder& product_holder) {
    return executors_.at(product_holder.product().absolute_norm()).get();
  }

 private:
  std::string name_;
  MarketExchangeApi mea_;
  OrderExecutorSystemConfig system_config_;
  std::map<std::string, std::unique_ptr<OrderExecutor>> executors_;  // order symbol => oe config
  IOrderSubsystem* os_;
  OrderQuerier* oq_;
};

}  // namespace impl

using impl::OrderExecutorSystem;  // for 1 mea

}  // namespace coin2::strategy::order_executor
